package fnic.prehand.esagent.ap_measure_new2;

import java.util.List;

import fnic.prehand.common.AgentConfig;
import fnic.river.RedisRiver;
import fnic.river.RedisSingleLoop;
 
public class APRedisRiver extends RedisRiver implements RedisSingleLoop, Runnable{
	private static final AgentConfig config = AgentConfig.getInstance();
	
	private static final String apflowKey = config.getApflowKey();
	private static final String staflowKey = config.getStaflowKey();
	private static final String staloginKey = config.getStaloginKey();
	private static final String stalogoutKey = config.getStalogoutKey();
	
	private List<String> response = null;
	private RecordDeal dealer;
	public APRedisRiver(String redisIp, int redisPort, String key) {
		super(redisIp, redisPort, key);
		dealer = new RecordDeal();
	}

	public void run() {
		// TODO Auto-generated method stub
		int i =0;
		while(true){
			setRedisKey(staloginKey);
			loopSingle();
	
//			int j = 0;
//			while(j<20){
			setRedisKey(apflowKey);
			loopSingle();
//			j++;
//			}
			dealer.doApBulk();
			
			
//			int k = 0;
//			while(k<10){
			setRedisKey(staflowKey);
			loopSingle();
//			k++;
//			}
			dealer.doStaBulk();
//			
//
//			
//			
			setRedisKey(stalogoutKey);
			loopSingle();
			
			i++;
		}
	}


	public void dealRecord(String item) {
		// TODO Auto-generated method stub
		if(redisKey == staflowKey){
			dealer.dealStaFlow(item);
		}else if(redisKey == staloginKey){
			dealer.dealStaLog(item, true);
		}else if(redisKey == stalogoutKey){
			dealer.dealStaLog(item, false);
		}else if(redisKey == apflowKey){
			dealer.dealApFlow(item);
		}
	}

	public void loopSingle() {
		int counter = 0;
		do {
			response = null;
			try {
				response = jedis.blpop(redisbulkTimeout, redisKey);
			} catch (Exception e) {
				// Can't get a redis object. Return and try again on th next loop.
				logger.info("Can't read from redis. Waiting 5 seconds and trying again.");
			}

			if (response != null) {
				String jsonsrc = response.get(1);
				dealRecord(jsonsrc);
			}
		} while (++counter < redisoncesize && response != null);
		cycle++;
	}
}
